package com.wasp.collector.service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.http.impl.client.CloseableHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.typesafe.config.ConfigFactory;
import com.wasp.collector.content.IContentHandler;
import com.wasp.collector.data.RequestContext;
import com.wasp.collector.data.RequestWrapper;
import com.wasp.collector.service.akka.ActorResult;
import com.wasp.collector.service.akka.CollectActorClient;
import com.wasp.collector.service.akka.ExceptionMonitor;
import com.wasp.collector.service.akka.HttpClientActorContext;
import com.wasp.collector.service.akka.INotify;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;

public class HttpClientExecutors implements ICollector,INotify {
	private static final long serialVersionUID = -2176118581539289848L;
	private static final Logger log = LoggerFactory.getLogger(HttpClientExecutors.class);

	private Object objLock = new Object();
	Map<Date, RequestContext> errors = Maps.newConcurrentMap();
	RequestWrapper requestWrapper = null;
	IContentHandler contentHandler = null;
	Set<RequestContext> nextUrls = Sets.newConcurrentHashSet();
	Map<String,ActorRef> registerActors = Maps.newConcurrentMap();
	boolean isBegin = false;
	ActorSystem system = null;
	List<RequestContext> requestContexts = new ArrayList<>();
	public HttpClientExecutors setRequest(RequestWrapper requestWrapper) {
		this.requestWrapper = requestWrapper;
		return this;
	}

	public HttpClientExecutors setContentHandler(IContentHandler contentHandler) {
		this.contentHandler = contentHandler;
		return this;
	}
	public HttpClientExecutors setRequestContexts (List<RequestContext> requestContexts){
		this.requestContexts = requestContexts;
		return this;
	}

	private void createActor(RequestContext rContext ,RequestWrapper wrapper) {
		
		/*ActorRef collectActorClient1 =
				system.actorOf(FromConfig.getInstance().props(Props.create(CollectActorClient.class)), 
				    "user/CollectActorClientRouter");
		
		ActorRef collectActorClient =
				system.actorOf(FromConfig.getInstance().props(Props.create(CollectActorClient.class)), 
				    "CollectActorClientRouter");*/
		
		ActorRef collectActorClient = system.actorOf(Props.create(CollectActorClient.class,this),
		        "client-"+UUID.randomUUID().toString());
		// 注册actor
		registerActors.put(rContext.getUrl(),collectActorClient);
		while(registerActors.size()>200){
			try {
				TimeUnit.SECONDS.sleep(10);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		collectActorClient.tell(new HttpClientActorContext(rContext,wrapper),null);
		/*Future<Object> f = Patterns.ask(collectActorClient, rContext, 10000000);
		onCompleteHandler(f);*/

	}

	@Override
	public void collect() {
		system = ActorSystem.create("ClusterSystem",ConfigFactory.load("stats1"));
		if (requestWrapper == null) {
			throw new RuntimeException("request should not be null");
		}
		if (contentHandler == null) {
			throw new RuntimeException("content handler should not be null");
		}
		iniMonitor();
		CollectorHelper.resgister(requestWrapper, contentHandler);
		HttpClientCollector collector = new HttpClientCollector(requestWrapper);
		CloseableHttpClient client = collector.build();
		try{
			isBegin = true;
			requestContexts.stream().forEach(
					item->visitUrl(item));
			while(true){
				//所有的请求都结束，则退出
				if(registerActors.isEmpty()){
					break;
				}
				/*try {
					TimeUnit.SECONDS.sleep(5000);
				} catch (InterruptedException e) {
					log.error("collect",e);
				}*/
				Set<RequestContext> nexts = Sets.newConcurrentHashSet();
				synchronized (objLock) {
					nexts = nextUrls.stream().collect(Collectors.toSet());
					nextUrls.clear();
				}
				if(!nexts.isEmpty()){
					nexts.stream().forEach(
							item->visitUrl(item));
				}
			}
		}catch(Exception e){
			log.error("collect",e);
		}finally{
			if(client !=null){
				try {
					client.close();
				} catch (IOException e) {
					log.error("IO", e);
				}
			}
		}
		
		
	}

	private void visitUrl( RequestContext rContext) {
		//HttpUriRequest request = new DoRequest(rContext, requestWrapper).create();
		createActor(rContext,requestWrapper);
	}

	private void onCompleteHandler(Object f) {
		if (f != null) {
			// actor出错
		}
		if (f instanceof ActorResult) {
			ActorResult result = (ActorResult) f;
			String url = result.getrContext().getUrl();
			registerActors.remove(url);
			if(result.getException() != null){
				monitorErrors(result);
				contentHandler.handleError(result);
				return;
			}
			List<RequestContext> nexts = result.getNexts();
			if(nexts!=null && !nexts.isEmpty()){
				synchronized (objLock) {
					nextUrls.addAll(nexts);
				}
			}
		}
	}
	public boolean isDone(){
		return isBegin && registerActors.isEmpty();
	}
	private void monitorErrors(ActorResult error){
		//重新请求error
		errors.put(error.getExecTime(), error.getrContext());
		visitUrl( error.getrContext());
	}

	@Override
	public Map<Date, RequestContext> getErrors() {
		return errors;
	}

	@Override
	public Map<String, ActorRef> getRegisterActors() {
		return registerActors;
	}
	
	private void iniMonitor(){
		ExceptionMonitor monitor = new ExceptionMonitor(this);
		monitor.start();
	}

	@Override
	public void completed(Object obj) {
		onCompleteHandler(obj);
		
	}
}